package com.sinozo.data.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.sinozo.data.common.ConfigConstant;
import com.sinozo.data.common.DataConfig;
import com.sinozo.data.utils.DateFormatUtil;
import com.sinozo.data.utils.MyKafkaUtil;
import com.sinozo.data.utils.PathUtil;
import com.sinozo.data.utils.ValidateUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;


/**
 * @Description: 用户行为日志分流
 * <p>
 * 1、一些必填字段的数据校验
 * 2、转化为JsonObject后进行分流
 * </p>
 * @Author: zhaoyunbao
 * @CreateDate: 2023/5/5 14:50
 * @Version: 1.0
 */
public class BaseLogApp {

    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(BaseLogApp.class);

        try {
            //1.获取执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(DataConfig.KAFKA_PARTITION_NUM);

            //1.1 开启CheckPoint
            env.enableCheckpointing(DataConfig.FLINK_CHECKPOINT_INTERVAL, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(DataConfig.FLINK_CHECKPOINT_TIMEOUT);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

            //1.2 设置状态后端
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
            );
            env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.of(3L, TimeUnit.DAYS), Time.of(1L, TimeUnit.MINUTES)));

            String hdfsPath = DataConfig.HDFS_SERVER;
            env.setStateBackend(new EmbeddedRocksDBStateBackend());
            env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(PathUtil.getHdfsPath(hdfsPath, "baseLogApp")));

            String topic = ConfigConstant.KAFKA_LOG_TOPIC;
            String groupId = ConfigConstant.KAFKA_LOG_GROUP_ID;
            String kafkaServer = DataConfig.KAFKA_SERVER;
            DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(kafkaServer, topic, groupId));

            //3.过滤掉非JSON格式的数据&将每行数据转换为JSON对象
            OutputTag<String> dirtyTag = new OutputTag<String>("Dirty") {
            };
            SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                    try {
                        JSONObject jsonObject = JSON.parseObject(value);
                        //校验数据格式
                        if (!ValidateUtils.validate(jsonObject)) {
                            ctx.output(dirtyTag, value);
                        } else {
                            out.collect(jsonObject);
                        }
                    } catch (Exception e) {
                        ctx.output(dirtyTag, value);
                        logger.error("json parse error", e);
                    }
                }
            });
            //获取侧输出流脏数据并打印
            DataStream<String> dirtyDS = jsonObjDS.getSideOutput(dirtyTag);
            dirtyDS.print("Dirty>>>>>>>>>>>>");

            // 4.按照设备(用户)分组
            KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("device_id"));

            // 5.新老访客状态标记修复
            SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(
                    new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                        ValueState<String> firstViewDtState;

                        @Override
                        public void open(Configuration param) throws Exception {
                            super.open(param);
                            firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>(
                                    "lastLoginDt", String.class
                            ));
                        }

                        @Override
                        public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {

                            Integer isNew = jsonObj.getJSONObject("common").getInteger("is_new");
                            String firstViewDt = firstViewDtState.value();
//                            Long ts = jsonObj.getLong("ts");
                            //替换ts为服务器系统时间
                            Long ts = System.currentTimeMillis();
                            jsonObj.put("ts", ts);
                            String dt = DateFormatUtil.toDate(ts);

                            if (isNew == 1) {
                                if (firstViewDt == null) {
                                    firstViewDtState.update(dt);
                                } else {
                                    if (!firstViewDt.equals(dt)) {
                                        isNew = 0;
                                        jsonObj.getJSONObject("common").put("is_new", isNew);
                                    }
                                }
                            } else {
                                if (firstViewDt == null) {
                                    // 将首次访问日期置为昨日
                                    String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                                    firstViewDtState.update(yesterday);
                                } else {
                                    if (firstViewDt.equals(dt)) {
                                        isNew = 1;
                                        jsonObj.getJSONObject("common").put("is_new", isNew);
                                    }
                                }
                            }
                            out.collect(jsonObj);
                        }
                    });

            // 6.使用侧输出流进行分流处理  页面日志放到主流  启动、动作、在线放到侧输出流
            OutputTag<String> startTag = new OutputTag<String>("start") {
            };
            OutputTag<String> actionTag = new OutputTag<String>("action") {
            };
            OutputTag<String> onlineTag = new OutputTag<String>("online") {
            };

            SingleOutputStreamOperator<String> pageDS = fixedStream.process(new ProcessFunction<JSONObject, String>() {
                @Override
                public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {

                    //获取公共信息&页面id&时间戳
                    JSONObject common = JSON.parseObject(value.getString("common"));
                    Long ts = value.getLong("ts");
                    //获取启动信息
                    String start = value.getString("start");
                    if (!StringUtils.isEmpty(start)) {
                        JSONObject startObj = JSON.parseObject(start);
                        JSONObject obj = new JSONObject();
                        obj.put("common", common);
                        obj.put("start", startObj);
                        obj.put("ts", ts);
                        ctx.output(startTag, obj.toJSONString());
                    }
                    //获取在线信息
                    String online = value.getString("online");
                    if (!StringUtils.isEmpty(online)) {
                        JSONObject onlineObj = JSON.parseObject(online);
                        JSONObject obj = new JSONObject();
                        obj.put("common", common);
                        obj.put("online", onlineObj);
                        obj.put("ts", ts);
                        ctx.output(onlineTag, obj.toJSONString());
                    }

                    //获取动作数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null && actions.size() > 0) {
                        //遍历数据&写到action侧输出流
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("ts", ts);

                            JSONObject obj = new JSONObject();
                            obj.put("action", action);
                            obj.put("common", common);
                            ctx.output(actionTag, obj.toJSONString());
                        }
                    }
                    //获取页面数据
                    JSONArray pages = value.getJSONArray("pages");
                    //移除动作数据&写到页面日志主流
                    value.remove("actions");
                    value.remove("pages");
                    value.remove("start");
                    value.remove("online");
                    if (pages != null && pages.size() > 0) {
                        for (int i = 0; i < pages.size(); i++) {
                            JSONObject page = pages.getJSONObject(i);
                            page.put("ts", ts);
                            value.put("page", page);
                            out.collect(value.toJSONString());
                        }
                    }
                }
//            }
            });

            // 7.提取各个侧输出流数据
            DataStream<String> startDS = pageDS.getSideOutput(startTag);
            DataStream<String> actionDS = pageDS.getSideOutput(actionTag);
            DataStream<String> onlineDS = pageDS.getSideOutput(onlineTag);

            //测试输出
            pageDS.print("page=>>>>>");
            startDS.print("start=>>>>>");
            actionDS.print("action=>>>>>");
            onlineDS.print("online=>>>>>");

            String page_topic = ConfigConstant.KAFKA_PAGE_LOG_TOPIC;
            String start_topic = ConfigConstant.KAFKA_START_LOG_TOPIC;
            String action_topic = ConfigConstant.KAFKA_ACTION_LOG_TOPIC;
            String online_topic = ConfigConstant.KAFKA_ONLINE_LOG_TOPIC;

            pageDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
            startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
            actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
            onlineDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(online_topic));

            //10.启动任务
            env.execute("BaseLogApp");
        } catch (Exception e) {
            logger.error("BaseLogApp error", e);
        }

    }

}
